package com.etl.component.basic;

import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.MapColumnDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 描述：
 *  列映射
 * @author xianggj
 * @Date 2021/10/12 18:10
 **/
public class ColumnTransformComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<MapColumnDTO> mapColumnDTOS =
                (List<MapColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        SetData input = (SetData)datas.get(0).getValue();
        DataSet<Row> dataset = (DataSet<Row>) input.getDataset();
        List<TableColumn> columns = input.getColumns();
        List<TableColumn> newColumns = getNewTableColumns(mapColumnDTOS, columns);
        setIndex(mapColumnDTOS, columns);
        //转换列类型
        RowTypeInfo rowTypeInfo = changeType(dataset, mapColumnDTOS);
        String callLocation = Utils.getCallLocationName();
        dataset = new MapOperator<Row, Row>(dataset, rowTypeInfo, e ->{
            for (MapColumnDTO mapColumnDTO  : mapColumnDTOS) {
                //类型不为空才进行转换
                if (StrUtil.isNotBlank(mapColumnDTO.getTarType())){
                    int index = mapColumnDTO.getIndex();
                    TypeInformation basicTypeInfo = typeToBasicTypeInfo(mapColumnDTO.getTarType());
                    Object field = e.getField(index);
                    if (field == null){
                        continue;
                    }
                    Object covert = null;
                    //转换值类型
                    try {
                        covert = covert(field, basicTypeInfo);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                    e.setField(index, covert);
                }
            }
            return e;
        }, callLocation);
        return execute(dataset, newColumns,"执行列映射操作");
    }

    /**
     * 类型转换
     * @param dataset
     * @param mapColumnDTOS
     * @return
     */
    protected RowTypeInfo changeType( DataSet<Row> dataset, List<MapColumnDTO> mapColumnDTOS) {
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        TypeInformation<?>[] fieldTypes = type.getFieldTypes();
        String[] fieldNames = type.getFieldNames();
        for (MapColumnDTO mapColumnDTO : mapColumnDTOS) {
            if (StrUtil.isNotBlank(mapColumnDTO.getTarType())){
                fieldTypes[mapColumnDTO.getIndex()] = typeToBasicTypeInfo(mapColumnDTO.getTarType());
            }
            fieldNames[mapColumnDTO.getIndex()] = mapColumnDTO.getTarField();
        }
        return new RowTypeInfo(fieldTypes, fieldNames);
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        ColumnData input = (ColumnData)datas.get(0).getValue();
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<MapColumnDTO> mapColumnDTOS =
                (List<MapColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        List<TableColumn> columns = input.getColumns();
        List<TableColumn> newColumns = getNewTableColumns(mapColumnDTOS, columns);
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(newColumns)));
    }

    /**
     * 获取新的字段类型
     * @param mapColumnDTOS
     * @param columns
     * @return
     */
    private List<TableColumn> getNewTableColumns(List<MapColumnDTO> mapColumnDTOS, List<TableColumn> columns) {
        Map<String, MapColumnDTO> filedMap = mapColumnDTOS.stream().collect(Collectors.toMap(MapColumnDTO::getField, Function.identity()));
        List<TableColumn> newColumns = new ArrayList<>();
        for (TableColumn column : columns) {
            TableColumn newColumn = column.clone();
            if (filedMap.containsKey(column.getColumnName())){
                MapColumnDTO mapColumnDTO = filedMap.get(column.getColumnName());
                newColumn.setColumnName(mapColumnDTO.getTarField());
                //原类型
                if (StrUtil.isNotBlank(mapColumnDTO.getTarType())){
                    newColumn.setDataType(mapColumnDTO.getTarType());
                    newColumn.setLength(mapColumnDTO.getTarLength());
                    newColumn.setFloatLength(mapColumnDTO.getTarecimal());
                }
            }
            newColumns.add(newColumn);
        }
        return newColumns;
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.COLUMN_TRANSFORM;
    }
}
